-module(xbt_tcp_callback).

-export([conn_entry/2]).

-include("xbt.hrl").

conn_entry(Socket,WSize)      ->
    case bind_user(Socket) of
        ok       ->
            conn_loop(Socket,WSize,0);
        Error    ->
            ?DEBUG("[~p]bind error: ~p~n",[self(),Error]),
            exit(Error)
    end.

bind_user(Socket)      ->
    ?DEBUG("[~p]try to bind user~n",[self()]),
    Pack = xbt_tcp_proto:get_packet(Socket),
    case xbt_tcp_proto:parse_packet(Pack) of
        {ok,{?TABLE_BIND,Seq,{User,AuthCode,_Version, TimeStamp}}} ->
            ?DEBUG("[~p]try to bind user, check user:~p~n",[self(),User]),
             case gen_server:call(xbt_access_server,{checkuser,{User,AuthCode,TimeStamp}}) of
                ok           ->
                    ?DEBUG("[~p]user ~p is ok, try to accept other commands~n",[self(),User]),
                    xbt_tcp_proto:send_packet(Socket,?TABLE_BIND_RESP,Seq,{?CODE_OK,?VERSION});
                _Error       ->
                    ok = xbt_tcp_proto:send_packet(Socket,?TABLE_BIND_RESP,Seq,{?CODE_BADCLIENT,?VERSION}),
                    ?DEBUG("[~p]gen_server:call error:~p~n",[self(),_Error]),
                    {error,bind}
            end;
        _                                                        ->
            {error,bind}
    end.


conn_loop(Socket,WSize,Num)  ->
    ?DEBUG("[~p]conn_loop: Socket=~p,WSize=~p,Num=~p~n",[self(),Socket,WSize,Num]),
    receive
        {error,peerClosed}    ->
            ?DEBUG("[~p]peer closed this connection,i'm exiting~n",[self()]),
            exit(peerClosed);
        {normal,unbind}        ->
            ?DEBUG("[~p]peer unbind,i'm exiting~n",[self()]),
            exit(normal);
        {error,badpacket}     ->
            ?DEBUG("[~p]got a bad packet~n",[self()]),
            case erlang:get(badpackets) of
                undefined            ->
                    erlang:put(badpackets,1),
                    conn_loop(Socket,WSize,Num);
                Num when Num < 3     ->
                    erlang:put(badpackets,Num+1),
                    conn_loop(Socket,WSize, Num);
                _                    ->
                    ?DEBUG("[~p]got too many badpackets,i'm exiting~n",[self()]),
                    exit(badPackets)
            end;
        {error,Other}        ->
            ?DEBUG("[~p]got error:~p, exiting~n",[self(),Other]),
            exit(Other);
        {'EXIT',_Pid,_Error} ->
            ?DEBUG("[~p]child worker: ~p exit for reason:~p, start a new one~n",[self(),_Pid,_Error]),
            conn_loop(Socket,WSize,Num-1)
    after
        0   ->
            if
                Num < WSize              ->
                    Pack = xbt_tcp_proto:get_packet(Socket),
                    start_worker(self(),Socket,Pack),
                    conn_loop(Socket,WSize,Num+1);
                true                     ->
                    %% sleep 10ms to avoid loop diretly when windows are full
                    ?DEBUG("[~p]running command windows is ~p(full)~n",[self(),Num]),
                    timer:sleep(10),
                    conn_loop(Socket,WSize,Num)
            end
    end.

start_worker(Parent,Socket,Pack)  ->
    spawn_link(fun() -> worker_entry(Parent,Socket,Pack) end).

worker_entry(Parent,Socket,Pack) ->
    case xbt_tcp_proto:parse_packet(Pack) of
        {ok,{Command,Sequence,Data}}    ->
            handle_request(Parent,Socket,{Command, Sequence,Data});
        {error,closed} ->
            %% close this connection.
            ?DEBUG("[~p]peer closed this connection,i'm exiting~n",[self()]),
            Parent ! {error,peerClosed};
        {error,badpacket} ->
            ?DEBUG("[~p]got bad packet~n",[self()]),
            Parent ! {error,badpacket};
        {error,Other}    ->
            %% close this connection.
            ?DEBUG("[~p]error: ~p~n",[self(),Other]),
            Parent ! {error,Other}
    end.

handle_request(Parent,Socket,{Command, Sequence,Data})  ->
    ?DEBUG("[~p]Got Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,Data]),
    case Command of
        ?TABLE_BIND                  ->
             Code = ?CODE_OK,
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{Code,?VERSION}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{?CODE_OK,?VERSION}]);
        ?TABLE_UNBIND                ->
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{}]),
             Parent ! {normal,unbind};
        ?TABLE_ACTIVE_TEST           ->
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{}]);
        ?TABLE_SET                   ->
             {Name,Value} = Data,
             Nodes = xbt_dht:find_nodes(Name,2),
             {Replies,BadNodes} = gen_server:multi_call(Nodes,
                 xbt_storage_server,{set,Name,Value},1500),
             ?DEBUG("[~p]set multi_call replies:~p,BadNodes:~p~n",[self(),Replies,BadNodes]),
             OkNodes = [ N || {N,ok} <- Replies ],
             if
                 length(Replies) == 0;length(BadNodes) > 0;
                 length(OkNodes) /= length(Nodes)            ->
                     Code = ?CODE_SERVERROR;
                 true                                        ->
                     Code = ?CODE_OK
             end,
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{Code}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{Code}]);
        ?TABLE_APPEND                ->
             {Name,Value} = Data,
             Nodes = xbt_dht:find_nodes(Name,2),
             {Replies,BadNodes} = gen_server:multi_call(Nodes,
                 xbt_storage_server,{append,Name,Value},1500),
             ?DEBUG("[~p]append multi_call replies:~p,BadNodes:~p~n",[self(),Replies,BadNodes]),
             OkNodes = [ N || {N,ok} <- Replies ],
             if
                 length(Replies) == 0;length(BadNodes) > 0;
                 length(OkNodes) /= length(Nodes)            ->
                     Code = ?CODE_SERVERROR;
                 true                                        ->
                     Code = ?CODE_OK
             end,
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{Code}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{Code}]);
        ?TABLE_DEL                   ->
             {Name} = Data,
             Nodes = xbt_dht:find_nodes(Name,2),
             {Replies,BadNodes} = gen_server:multi_call(Nodes,
                  xbt_storage_server,{delete,Name},1500),
             ?DEBUG("[~p]delete multi_call replies:~p,BadNodes:~p~n",[self(),Replies,BadNodes]),
             OkNodes = [ N || {N,ok} <- Replies ],
             if
                 length(Replies) == 0;length(BadNodes) > 0;
                 length(OkNodes) /= length(Nodes)            ->
                     Code = ?CODE_SERVERROR;
                 true                                        ->
                     Code = ?CODE_OK
             end,
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{Code,Name}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{Code,Name}]);
        ?TABLE_GET                   ->
             {Name} = Data,
             Nodes = xbt_dht:find_nodes(Name,2),
             {Replies,_BadNodes} = gen_server:multi_call(Nodes,
                  xbt_storage_server,{get,Name},1500),
             ?DEBUG("[~p]get multi_call replies:~p,BadNodes:~p~n",[self(),Replies,_BadNodes]),
             Ok = [ V || {_,{ok,V}} <- Replies ],
             NO = [ N || {N,{error,undefine}} <- Replies ],
             if
                 length(Ok) > 0               ->
                     Code = ?CODE_OK,
                     [Value|_Tail] = Ok;
                 length(NO) > 0               ->
                     Code = ?CODE_NOKEY,
                     Value = <<>>;
                 true                                        ->
                     Code = ?CODE_SERVERROR,
                     Value = <<>>
             end,
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{Code,Name,Value}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{Code,Name,Value}]),
             ResetNodes = [ N || {N,{error,_}} <- Replies ],
             if
                 length(Ok) > 0,length(ResetNodes) > 0  ->
                     ?DEBUG("[~p]reset failed nodes after get:~p~n",[self(),ResetNodes]),
                     gen_server:multi_call(ResetNodes,
                          xbt_storage_server,{set,Name,Value},1500);
                 true                                   ->
                     ok
             end;
        _                            ->
             ok = xbt_tcp_proto:send_packet(Socket,Command+1,Sequence,{}),
             ?DEBUG("[~p]Reply Command:~p,Sequence:~p,Data:~p:~n", [self(),Command,Sequence,{}])
    end.
